package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * Created by 黄凯 on 2023/7/5 0005 11:27
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * 对关联后的数据进行处理，过滤出维度数据传递到下游
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Map<String, TableProcess> configMap = new HashMap<>();

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        //在对两条流数据进行处理之前，将配置表中的配置信息提前预加载到程序的Map中
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");

        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall0201_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");

        //获取数据库操作对象
        String sql = "select * from gmall0201_config.table_process_dim";
        PreparedStatement ps = conn.prepareStatement(sql);

        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();

        //处理结果集
        while (rs.next()) {
            //定义一个json对象，用于接收遍历出来的一行数据
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName, columnValue);
            }
            //将jsonObj转换为实体类对象
            TableProcess tableProcess = JSON.toJavaObject(jsonObj, TableProcess.class);
            //将读取的配置信息放到configMap集合中
            configMap.put(tableProcess.getSourceTable(), tableProcess);
        }

        //释放资源
        rs.close();
        ps.close();
        conn.close();

    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj,
                               BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.ReadOnlyContext readOnlyContext,
                               Collector<JSONObject> collector) throws Exception {

        // {"database":"gmall0201","data":{"tm_name":"Redmi11","create_time":"2021-12-14 00:00:00","logo_url":"/aaa/bbb","id":1},"old":{"tm_name":"Redmi"},"type":"update","table":"base_trademark","ts":1688523334}
        // System.out.println(jsonObj);
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);

        //获取当前处理业务数据库表的表名
        String key = jsonObj.getString("table");

        //根据表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;

        //如果配置信息不为空，说明是维度，继续传递到下游（只需要将data内容传递即可）
        if ((tableProcess = broadcastState.get(key)) != null
                || (tableProcess = configMap.get(key)) != null) {

            // {"tm_name":"Redmi11333","create_time":"2021-12-14 00:00:00","logo_url":"/aaa/bbb","id":1}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //在向下游传递数据前，过滤掉不需要传递的字段
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj, sinkColumns);

            //在向下游传递数据前，补充操作类型属性
            // {"tm_name":"Redmi11333","create_time":"2021-12-14 00:00:00","logo_url":"/aaa/bbb","id":1,"type":"update"}
            String type = jsonObj.getString("type");
            dataJsonObj.put("type", type);

            //在向下游传递数据前，补充配置信息
            dataJsonObj.put("table_process", tableProcess);


            collector.collect(dataJsonObj);

        }

    }

    //过滤掉不需要传递到下游的字段
    //dataJsonObj: {"tm_name":"Redmi11333","create_time":"2021-12-14 00:00:00","logo_url":"/aaa/bbb","id":1}
    //sinkColumns: id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {

        String[] columnArr = sinkColumns.split(",");
        Set<String> columnSet = new HashSet<>(Arrays.asList(columnArr));

        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();

        entrySet.removeIf(entry -> !columnSet.contains(entry.getKey()));

    }

    //处理广播流配置数据
    @Override
    public void processBroadcastElement(TableProcess tableProcess,
                                        BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context context,
                                        Collector<JSONObject> collector) throws Exception {

        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);

        //获取对配置表操作类型
        String op = tableProcess.getOp();

        //获取维度表表名
        String key = tableProcess.getSourceTable();

        //判断是否为删除操作
        if ("d".equals(op)) {

            //如果是删除操作，从广播状态中删除掉对应的配置信息
            broadcastState.remove(key);

            configMap.remove(key);

        } else {

            //如果是删除外的其它操作，将配置信息更新到广播状态中
            broadcastState.put(key, tableProcess);

            configMap.put(key, tableProcess);

        }

    }
}
